feat(data-plane): read query-time raw windows from a local raw buffer and Prometheus remote read - #831

…puts

Each MatAtIngestion output the query side reads (the executor's
materialization frontier) reads one tumbling pane of raw samples. Outputs
that differ only in which pane they read share one SummaryDefinition,
identified by a SHA-256 of their sub-DAG with node ids blanked and offset
TimeShifts skipped. Each output binds to its definition with an offset: at
evaluation time T it reads the pane ending at T - offset. SummaryStore keeps
built panes by definition and end. A plan with a node kept across
evaluations is rejected: the backend declares it cannot keep one.

#509 Example 4b on a deployment without raw data: its six 10-minute KLL
panes become one definition with offsets 0..50 min.

An engine thread owns the installed plan (Planner IR holds Rcs) and receives
Remote Write samples from the HTTP server. For each stored summary
definition it buffers its Scan metric's samples by tumbling pane
(end - width, end], like PromQL ranges. When the watermark (newest sample
minus --allowed-lateness-ms) passes a pane's end, it runs the plan's Scan
over them (applying label matchers) and the definition's build from its
TimeRange input, then stores the output. Panes without samples are stored
empty; the first pane seen is skipped because it may be partial; samples for
built panes are dropped and counted. A full engine queue answers Remote
Write with 503 so senders retry.

Each plan root is compiled once with two kinds of inputs: stored outputs,
bound at evaluation time T to their definition's pane ending at T - offset,
and query-time raw ranges (TimeRange <- TimeShift* <- Scan), bound to the
metric's samples in (T - offset - range, T - offset] fetched from
Prometheus and filtered by the plan's Scan. An instant query whose text is a
planned query is answered with a Prometheus vector and x-asap-execution:
plan. Unplanned queries are forwarded; planned ones that cannot be answered
(pane missing or off the pane grid, fetch or execution failure) are
forwarded with x-asap-forward-reason.

For Examples 1, 3a, 3b, 4a and 4b: plan the workload, start the data_plane
binary against a Prometheus holding generated samples, Remote Write the
same samples, and require every planned query to be answered by the plan
with the Planner executor's result for the selected logical candidate with
nothing materialized (control_plane::plan_selected_at_query_time). 3b also
runs, and 4b only runs, on a deployment without raw data, so their panes are
built from Remote Write. Exact results must match exactly; Example 3a/4a's
five-year KLL windows exceed k and KLL compaction is randomized, so they
must agree within twice the 0.005 rank target.

A plan node materialized `MatAtQueryKeptAcrossEvaluations` (#509 Example 4's
B3) is now executed instead of rejected. Its output is kept in memory by
(definition, pane end), with the SDS identity rule, built from its pane's raw
samples the first time an evaluation reads it, reused by later evaluations,
and evicted like stored panes (older than the largest read offset plus two
pane widths, measured from the newest evaluation).

The control plane's default DeploymentCapabilities now set
query_time_results_retained. examples/planner-layering-example4-b3.json is a
workload for which the Planner selects B3 under the default cost model.

A pane computed at query time is kept across evaluations only once its end
is at or before the precompute engine's watermark (newest sample minus the
allowed lateness). A pane read while still open is computed for that
evaluation but not kept, so a later evaluation computes it again, including
samples that arrived late, and keeps it once it is closed.

A planned /api/v1/query_range runs one evaluation per step, start + k*step
through end, and returns a Prometheus matrix. Each query-time raw range is
fetched from Prometheus once for all steps and sliced per step; each kept
definition's missing panes are fetched once, so a kept pane built at one
step is reused by the later ones. Stored panes are read per step. If any
step cannot be answered, the whole request is forwarded with the reason.

start/end accept Unix seconds or RFC 3339 (now also for instant queries),
step float seconds or a Prometheus duration; requests Prometheus would
reject, and requests with a nonzero limit, are forwarded unchanged.

…and asap-executor only

Re-pin ASAPPlanner to main f0dc1825 and use its handoff API.

- The control plane calls `asap_planner::plan_installable` once and writes
  the returned `InstallablePlan` as plan.json. The backend's own pane-chain
  expansion, Stage 2/3 plumbing and query-time reference rebuild are gone;
  tests use `InstallablePlan::query_time_reference`.
- The data plane reads `stored_outputs` (built_at, pane window,
  retention_ms) and `query_time_raw_windows`, and selects every pane's and
  raw window's rows with `asap_executor::RawSampleSelector`. It no longer
  uses `frontier_from_timing` or `promql_rows`.
- The SDS keeps its own identity (sub-DAG hash, offset-only TimeShifts
  transparent) and keeps each definition for the largest `retention_ms`
  among its outputs, measured from the newest evaluation that can read it,
  instead of "largest offset plus two pane widths".
- Among Planner crates the workspace depends on `asap-planner` and
  `asap-executor` only; IR types come through `asap_executor::planner`.
- `--plan-selection-settings` on the control plane and
  `--execution-memory-limit-mib` on the data plane, which the benefit
  harness passes.

…a shared pane store

- Ingestion: N workers (`--ingestion-workers`, default one per core).
  Remote Write series are sharded by a hash of their label set; each worker
  owns its shard's open panes. The global watermark is the minimum over the
  workers that have seen a sample; when it passes a pane, every worker
  builds its part, and the last part completes the pane: parts are joined
  and the states of a group several workers saw are merged with the
  family's merge. A request is queued to every worker or refused whole.
- Queries: a pool of `--query-threads` threads, one query (and every step
  of a range query) per thread. The executor's compiled plans hold `Rc`s,
  so each thread reads plan.json and compiles its own.
- SDS: `RwLock<HashMap<(definition, pane end), Arc<pane>>>`. Panes are
  inserted once and never rewritten; readers clone the `Arc`. Kept
  query-time results use the same store, shared by the query threads.
  A stored pane binds to a root as a shared source, copied once per run
  (the executor's sources emit owned batches) instead of twice.
- jemalloc is the data plane's global allocator.

…he watermark it announces

Two races of the sharded engine, both found by a test that writes a pane's
samples and queries it at once:

- A planned query ran on the query pool while the ingestion workers could
  still be ingesting the Remote Write requests answered before it, so the
  newest pane was missing and the query was forwarded. A query now waits
  until every worker has run what was queued before it (`sync`), as the
  single engine thread's queue ordered it.
- A worker closed its panes at the watermark it read before another worker
  moved it, then announced the newer watermark to the other workers but
  not to itself, so its part of the newest pane never arrived. It now
  closes and announces the same watermark.

The ingestion side of a stored definition, Scan -> TimeShift* ->
TimeRange(pane) -> SummaryAgg, runs per sample instead of over each pane's
buffered rows:

- per series, once: the executor's RawSampleSelector turns the series'
  first sample into its input row (applying the Scan's matchers), and the
  row's group columns give its group; cached by label set;
- per sample: the open pane's state for that group is updated in place
  with the updater the executor's batch build uses for the family
  (create_planner_accumulator), with the same value and timestamp;
- per group, once: the definition's compiled build runs over the group's
  first row, and its output row is the template each closed pane's row is
  made from (pane timestamp and state replaced).

A build of another shape (row filter, keyed or shared-grid family,
operators between the range and the summary, matchers reading the value or
time) keeps the batch build. A parity test checks every family the
examples store (counter rate, sum, KLL at two pane widths) against the
executor's batch build of the same samples: exact states byte for byte,
KLL by quantile (KLL seeds its compaction randomly per sketch).

… and Prometheus remote read

- A local raw buffer keeps the recent samples Remote Write delivered,
  sharded like ingestion: each ingestion worker appends its shard. It keeps
  the plan's longest query-time raw read back from the newest sample: the
  largest `end_offset + width` of `query_time_raw_windows` and of the panes
  built at query time and kept (their retention, which covers range
  evaluations). `--raw-buffer-ms` overrides it; 0 keeps none.
- A raw window is read from the buffer; the part before the buffer's cut
  (its first sample, or the trimmed horizon of any shard) is read with
  Prometheus remote read (protobuf, `/api/v1/read`). The JSON query API is
  no longer used for raw samples. Forwarding of unplanned queries is
  unchanged.
- The data plane logs the buffer's series and samples every 30 s.

…est query-time read

The buffer kept every metric for the plan's single longest raw read: in the
benefit plan, 60 minutes of the counter whose raw reads span one minute,
and of metrics no raw window reads. It now keeps only the metrics the plan
reads at query time, each for its own longest read (query-time raw windows,
and panes built at query time and kept).